import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, run}
import com.cw.realtime.common.bean.CartAddUniqueBean
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.DateFormatUtil.TsSyntax
import com.cw.realtime.common.util.FlinkSinkUtil
import io.circe._
import io.circe.syntax._
import io.circe.generic.auto._
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration


object TradeCartAdd {

  def main(args: Array[String]): Unit = {
    implicit val conf = RunnerConfig("dws_trade_cart_add_uu_window", TOPIC_DWD_TRADE_CART_ADD, 10026, ckp = true)
    run { (_, ds) =>
      // 数据清洗
      val jsonObjStream = etl(ds)

      // 水位线
      val jsonObjectStreamWM = withWatermark(jsonObjStream)

      // 分组判断独立用户
      val cartStream = judgeUnique(jsonObjectStreamWM)

      // 开窗聚合
      val aggStream = windowAggregate(cartStream)

      aggStream.print()
      // sink
      aggStream
        .map(_.asJson.noSpaces)
        .sinkTo(FlinkSinkUtil.getDorisSink(DWS_TRADE_CART_ADD_UU_WINDOW))



    }


    def windowAggregate(ds: DataStream[CartAddUniqueBean]) = {
      ds.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
        .reduce(_ + _, new AllWindowFunction[CartAddUniqueBean, CartAddUniqueBean, TimeWindow] {
          override def apply(window: TimeWindow, input: Iterable[CartAddUniqueBean], out: Collector[CartAddUniqueBean]): Unit = {
            import com.cw.realtime.common.util.TimeWindowUtil._
            val (stt, edt, curDate) = window.windowRange
            input.foreach { e => out.collect(e.transWindowRange(stt, edt, curDate)) }
          }
        })


    }


    def judgeUnique(ds: DataStream[JsonObject]) = {
      ds
        .keyBy(obj => obj("user_id").get.asString.get)
        .process(new KeyedProcessFunction[String, JsonObject, CartAddUniqueBean] {
          type FuncT = KeyedProcessFunction[String, JsonObject, CartAddUniqueBean]

          val desc =
            new ValueStateDescriptor("add_date", classOf[String])

          var state = Option.empty[ValueState[String]]

          override def open(parameters: Configuration): Unit = {
            desc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build())
            state = Some(getRuntimeContext.getState(desc))
          }

          override def processElement(value: JsonObject, ctx: FuncT#Context, out: Collector[CartAddUniqueBean]): Unit = {
            for {
              dateState <- state
              ts <- value("ts").get.asNumber.get.toLong
            } {
              val lastDate = dateState.value
              val curDate = ts.toDate
              if (lastDate == null || curDate != lastDate) {
                dateState.update(curDate)
                out.collect(CartAddUniqueBean(1))
              }
            }
          }
        })


    }


    def etl(ds: DataStream[String]) = {
      ds.flatMap { s =>
        for {
          json <- parser.parse(s).toOption
          jsonObj <- json.asObject
          _ <- jsonObj("user_id")
          ts <- jsonObj("ts")
          ts <- ts.asNumber
          ts <- ts.toLong
        } yield jsonObj.add("ts", (ts * 1000).asJson)
      }
    }


    def withWatermark(ds: DataStream[JsonObject]) = {
      ds.assignTimestampsAndWatermarks(
        WatermarkStrategy.forBoundedOutOfOrderness(
          Duration.ofSeconds(3)).withTimestampAssigner(
          new SerializableTimestampAssigner[JsonObject] {
            override def extractTimestamp(element: JsonObject, recordTimestamp: Long): Long =
              element("ts").get.asNumber.get.toLong.get
          }))


    }


  }


}
